package cn.hy.dataSync.transport.channel;


import cn.hy.dataSync.config.CoreConfig;
import cn.hy.dataSync.config.CoreConstant;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import cn.hy.dataSync.transport.record.TerminateRecord;
import cn.hy.dataSync.utils.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.Validate;

import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


@Slf4j
public abstract class Channel {

    protected int taskId;

    protected int capacity;

    protected long recordSpeed;

    protected long flowControlInterval;

    protected volatile boolean isClosed = false;

    protected CoreConfig coreConfig;

    protected AtomicLong waitReaderTime = new AtomicLong(0);

    protected AtomicLong waitWriterTime = new AtomicLong(0);

    private Communication currentCommunication;

    private Communication lastCommunication = new Communication();

    public Channel(TaskConfig taskConfig) {
        this.coreConfig = taskConfig.getCoreConfig();
        this.capacity = coreConfig.getInteger(CoreConstant.CHANNEL_CAPACITY, 2048);
        if (this.capacity <= 0) {
            throw new IllegalArgumentException(String.format("通道容量[%d]必须大于0.", this.capacity));
        }
        this.recordSpeed = taskConfig.getJobConfig().getSpeedConfig().getRecord();
        if (this.recordSpeed <= 0) {
            this.recordSpeed = coreConfig.getLong(CoreConstant.CHANNEL_SPEED_RECORD, 10000);
        }
        this.flowControlInterval = coreConfig.getLong(CoreConstant.FLOW_CONTROL_INTERVAL, 1000);
    }

    public void close() {
        this.isClosed = true;
    }

    public void open() {
        this.isClosed = false;
    }

    public boolean isClosed() {
        return isClosed;
    }

    public int getTaskId() {
        return this.taskId;
    }

    public int getCapacity() {
        return capacity;
    }

    public CoreConfig getCoreConfig() {
        return this.coreConfig;
    }

    public void setCommunication(final Communication communication) {
        this.currentCommunication = communication;
        this.lastCommunication.reset();
    }

    public void push(final Record r) {
        Validate.notNull(r, "record不能为空.");
        this.doPush(r);
        this.statPush(1L);
    }

    public void pushTerminate(final TerminateRecord r) {
        Validate.notNull(r, "record不能为空.");
        this.doPush(r);
    }

    public void pushAll(final Collection<Record> rs) {
        Validate.notNull(rs);
        Validate.noNullElements(rs);
        this.doPushAll(rs);
        // 记录
        this.statPush(rs.size());
    }

    public Record pull() {
        Record record = this.doPull();
        this.statPull(1L);
        return record;
    }

    public void pullAll(final Collection<Record> rs) {
        Validate.notNull(rs);
        this.doPullAll(rs);
        this.statPull(rs.size());
    }

    /**
     * 单条记录推送
     *
     * @param r 记录
     */
    protected abstract void doPush(Record r);

    /**
     * 批量记录推送
     *
     * @param rs 批量记录
     */
    protected abstract void doPushAll(Collection<Record> rs);

    /**
     * 单条记录拉取
     *
     * @return 记录
     */
    protected abstract Record doPull();

    /**
     * 批量记录拉取
     *
     * @param rs 批量记录
     */
    protected abstract void doPullAll(Collection<Record> rs);

    public abstract int size();

    public abstract boolean isEmpty();

    public abstract void clear();

    private void statPush(long recordSize) {
        currentCommunication.increaseCounter(CommunicationTool.READ_SUCCEED_RECORDS, recordSize);
        currentCommunication.setLongCounter(CommunicationTool.WAIT_READER_TIME, waitReaderTime.get());
        currentCommunication.setLongCounter(CommunicationTool.WAIT_WRITER_TIME, waitWriterTime.get());
        boolean isChannelRecordSpeedLimit = (this.recordSpeed > 0);
        if (!isChannelRecordSpeedLimit) {
            return;
        }
        long lastTimestamp = lastCommunication.getTimestamp();
        long nowTimestamp = System.currentTimeMillis();
        long interval = nowTimestamp - lastTimestamp;
        if (interval - this.flowControlInterval >= 0) {
            long recordLimitSleepTime = 0;

            long currentRecordSpeed = (CommunicationTool.getTotalReadRecords(currentCommunication) -
                    CommunicationTool.getTotalReadRecords(lastCommunication)) * 1000 / interval;
            if (currentRecordSpeed > this.recordSpeed) {
                // 计算根据recordLimit得到的休眠时间
                recordLimitSleepTime = currentRecordSpeed * interval / this.recordSpeed - interval;
            }

            if (recordLimitSleepTime > 0) {
                TimeUtils.sleepUninterruptibly(recordLimitSleepTime, TimeUnit.MILLISECONDS);
            }

            lastCommunication.setLongCounter(
                    CommunicationTool.READ_SUCCEED_RECORDS,
                    currentCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS));
            lastCommunication.setLongCounter(
                    CommunicationTool.READ_FAILED_RECORDS,
                    currentCommunication.getLongCounter(CommunicationTool.READ_FAILED_RECORDS));
            lastCommunication.setTimestamp(nowTimestamp);
        }
    }

    private void statPull(long recordSize) {
        currentCommunication.increaseCounter(CommunicationTool.WRITE_RECEIVED_RECORDS, recordSize);
    }
}
